package com.atguigu.sparkstreaming.demos

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * Created by Smexy on 2022/8/22
 *

 *
 *
 *
 */
object TransformDemo {

  def main(args: Array[String]): Unit = {


    val streamingContext = new StreamingContext("local[*]", "wordcount", Seconds(5))

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "sz220409test",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> "true"
    )


    val topics = Array("topicD")


    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    val ds1: DStream[(String, Int)] = ds.flatMap(record => record.value().split(" "))
      .map(word => (word, 1))
      .reduceByKey(_ + _)

    /*
        获取每个单词，按照单词的key进行排序后再输出

        sortbykey这个算子在 DStream中没有，但是在RDD中有，此时可以将对 DStream的运算转换为对DStream中所包含的RDD的运算。
     */
    val ds2: DStream[(String, Int)] = ds1.transform(rdd => {

      val rdd1: RDD[(String, Int)] = rdd.sortByKey()
      rdd1

    })

    //⑤输出结果
    ds2.print(1000)

    //⑥启动App
    streamingContext.start()

    //等待发送终止命令，不发送，阻塞在这行，一直等
    streamingContext.awaitTermination()

  }

}
